package spider.zookeeper.node;

import org.apache.zookeeper.CreateMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import spider.zookeeper.BaseZKClient;

/**
 * 节点管理
 *
 * @author Archx[archx@foxmail.com]
 * @date 2017/4/7.
 */
public class NodeManager extends BaseZKClient {

    private static final Logger logger = LoggerFactory.getLogger(NodeManager.class);

    /**
     * 根节点
     */
    private String rootPath;

    /**
     * 节点标识
     */
    private String nodeId;

    /**
     * 节点链接地址
     */
    private String nodeUrl;

    /**
     * 繁忙节点
     */
    private String busyPath;

    /**
     * 阈值
     */
    private int threshold;

    /**
     * 繁忙标识
     */
    private boolean busy;

    @Override
    public void execute() {
        logger.info("> Sync zookeeper: [ path = {}/{}]", rootPath, nodeId);
        // 创建节点
        createNode(rootPath + "/" + nodeId, nodeUrl.getBytes(), CreateMode.EPHEMERAL);
    }

    /**
     * 获取访问路径
     *
     * @param nodeId 节点标识
     * @return
     */
    public String getNodeUrl(String nodeId) {
        return getNodeData(rootPath + "/" + nodeId);
    }

    /**
     * 更新繁忙标识
     *
     * @param conn 连接数
     */
    public void updateBusy(int conn) {
        if (conn >= threshold) {
            if (!busy) {
                logger.info("> [ NodeId = {} ] busy!!", nodeId);
                createNode(busyPath + "/" + nodeId, new byte[0], CreateMode.EPHEMERAL);
                busy = true;
            }
        } else {
            if (busy) {
                logger.info("> [ NodeId = {} ] available again!!", nodeId);
                removeNode(busyPath + "/" + nodeId);
            }
            busy = false;
        }
    }

    public void setRootPath(String rootPath) {
        this.rootPath = rootPath;
    }

    public void setNodeId(String nodeId) {
        this.nodeId = nodeId;
    }

    public void setNodeUrl(String nodeUrl) {
        this.nodeUrl = nodeUrl;
    }

    public void setBusyPath(String busyPath) {
        this.busyPath = busyPath;
    }

    public void setThreshold(int threshold) {
        this.threshold = threshold;
    }

    public void setBusy(boolean busy) {
        this.busy = busy;
    }
}
